package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.util.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/22
 *
     实时统计每小时内的网站UV
            基于EventTime 时间的滚动窗口。窗口的范围(大小)是1h

        window的效果:
        ------------------------
                a,10:01,pv
                a,10:02,pv
                a,10:03,pv
                    --------->Process1    10:1 (a)

        -------------------------
                 b,10:01,pv
                 b,10:02,pv
                 b,10:03,pv
                 --------->Process2    10:1 (b)

              结果:  10:2  (a,b)

        -------------------
            全局汇总的场景，保证，所有用户的PV记录，得处于同一个Process，才能全局汇总。


        ------------------
             windowAll:
                        开窗，全局一个process。
                        这个Process的key是固定的
                            <A,UserBehavior(a)>
                            <A,UserBehavior(b)>
                            <A,UserBehavior(c)>
                                    不管你是什么时间窗口下的数据，都会
                                    共享同一个    private MapState<Long, Object> uids;
 */
public class Demo2_UV
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((u, t) -> u.getTimestamp() * 1000);

        env.readTextFile("data/UserBehavior.csv")
           .flatMap(new FlatMapFunction<String, UserBehavior>()
           {
               @Override
               public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                   String[] words = value.split(",");
                   if ("pv".equals(words[3])){
                       out.collect(new UserBehavior(
                            Long.valueOf(words[0]),
                            Long.valueOf(words[1]),
                            Integer.valueOf(words[2]),
                           words[3],
                            Long.valueOf(words[4])
                       ));
                   }
               }
           })
           .assignTimestampsAndWatermarks(watermarkStrategy)
           //把同一个用户的PV记录分到一起
           //.keyBy(UserBehavior::getUserId)
           .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
           .process(new ProcessAllWindowFunction<UserBehavior, String, TimeWindow>()
           {

               //用来去重
               private MapState<Long, Object> uids;

               @Override
               public void open(Configuration parameters) throws Exception {
                   uids = getRuntimeContext().getMapState(new MapStateDescriptor<Long, Object>("uids", Long.class, Object.class));
               }

               @Override
               public void process(Context context, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {

                   //清空MapState，保证此次运行只有当前窗口的数据统计
                   uids.clear();

                   //把1h收到的数据加入到状态，进行去重
                   for (UserBehavior userBehavior : elements) {
                       uids.put(userBehavior.getUserId(),"a");
                   }

                   //获取到去重的人数
                   out.collect(MyUtil.parseTimeWindow(context.window()) + ":"+MyUtil.parseList(uids.keys()).size());

               }
           })
           .print().setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
